package com.univocity.trader.exchange.interactivebrokers.api;

import com.ib.client.*;
import org.apache.commons.lang3.*;
import org.slf4j.*;

import java.util.*;
import java.util.function.*;

/**
 * @author uniVocity Software Pty Ltd - <a href="mailto:dev@univocity.com">dev@univocity.com</a>
 */
abstract class IBRequests {

	private static final Logger log = LoggerFactory.getLogger(IBRequests.class);

	protected EClientSocket client;
	private EJavaSignal signal;
	private EReader reader;

	final RequestHandler requestHandler;
	final ResponseProcessor responseProcessor;


	final String ip;
	final int port;
	final int clientID;
	final String optionalCapabilities;
	private boolean reconnecting = false;

	public IBRequests(String ip, int port, int clientID, String optionalCapabilities, Runnable reconnectionProcess) {
		this.ip = ip;
		this.port = port;
		this.clientID = clientID;
		this.optionalCapabilities = optionalCapabilities;

		this.requestHandler = new RequestHandler(reconnectionProcess);
		this.responseProcessor = new ResponseProcessor(requestHandler);
	}

	abstract IBRequests newInstance(IBRequests oldInstance);

	private EJavaSignal getSignal() {
		if (signal == null) {
			synchronized (this) {
				if (signal == null) {
					signal = new EJavaSignal();
				}
			}
		}
		return signal;
	}

	private EClientSocket getClient() {
		if (client == null) {
			synchronized (this) {
				if (client == null) {
					signal = null;
					client = new EClientSocket(responseProcessor, getSignal());
					client.optionalCapabilities(optionalCapabilities);
					client.eConnect(ip, port, clientID);
					if (client.isConnected()) {
						log.info("Connected to TWS server (version {}})", client.serverVersion());
					} else {
						client = null;
						throw new IllegalStateException("Could not connect to TWS. Make sure it's running on " + (StringUtils.isBlank(ip) ? "localhost" : ip) + ":" + port);
					}
				}
			}
		}
		return client;
	}

	public static IBRequests reconnect(IBRequests ibRequests) {
		if (ibRequests != null && !ibRequests.reconnecting) {
			synchronized (IBRequests.class) {
				if (!ibRequests.reconnecting) {
					ibRequests.reconnecting = true;
					return ibRequests.newInstance(ibRequests);
				}
			}
		}
		return ibRequests;
	}

	private EReader getReader() {
		if (reader == null) {
			synchronized (this) {
				if (reader == null) {
					log.info("Connecting to IB...");
					reader = new EReader(getClient(), getSignal());
					reader.start();
				}
			}
		}
		return reader;
	}

	synchronized void connect() {
		boolean[] ready = new boolean[]{false};

		EReader reader = null;
		int retryCount = 0;
		while (reader == null && ++retryCount <= 10) {
			try {
				reader = getReader();
			} catch (RuntimeException e) {
				if (retryCount == 10) {
					log.error("Unable to connect after " + retryCount + " attempts. Aborting.", e);
					throw e;
				}
				log.error("Unable to connect. Will retry in 30 seconds", e);
				try {
					Thread.sleep(30_000);
				} catch (InterruptedException ie) {
					Thread.currentThread().interrupt();
				}
			}
		}

		new Thread(() -> {
			Thread.currentThread().setName("IB live stream");
			while (getClient().isConnected()) {
				ready[0] = true;
				getSignal().waitForSignal();
				try {
					getReader().processMsgs();
				} catch (Exception e) {
					log.error("Error processing messages", e);
				}
			}
			log.warn("IB live stream stopped");
		}).start();

		while (!ready[0]) {
			Thread.yield();
		}
		log.info("Connected to TWS.");
	}

	public synchronized void disconnect() {
		if (client != null) {
			log.warn("Disconnecting from IB live stream");
			client.cancelRealTimeBars(3001);

			if (signal != null) {
				try {
					signal.issueSignal();
				} catch (Exception e) {
					//ignore. Don't care.
				}
			}

			if (reader != null) {
				try {
					reader.interrupt();
					reader = null;
				} catch (Exception e) {
					//ignore. Don't care.
				}
			}

			try {
				client.eDisconnect();
			} catch (Exception e) {
				//ignore. Don't care.
			}

			client = null;
			signal = null;
		}
	}

	protected final <T> int submitRequest(String description, Consumer<T> resultConsumer, Consumer<Integer> action) {
		return submitRequest(0, description, resultConsumer, action);
	}

	protected final <T> int submitRequest(int requestId, String description, Consumer<T> resultConsumer, Consumer<Integer> action) {
		requestId = requestHandler.prepareRequest(requestId, resultConsumer);
		if (requestId != 0) {
			log.debug("New request [" + requestId + "]: " + description);
			action.accept(requestId);
		}
		return requestId;
	}

	public void waitForResponse(int requestId) {
		if (requestId == 0) {
			log.warn("Previous request was ignored as a similar one is already being processed");
			return;
		}
		waitForResponse(requestId, 10);
	}

	public void waitForResponse(int requestId, int maxSecondsToWait) {
		requestHandler.waitForResponse(requestId, maxSecondsToWait);
	}

	public void waitForResponses(Collection<Integer> requestIds) {
		waitForResponses(requestIds, 10);
	}

	public void waitForResponses(Collection<Integer> requestIds, int maxSecondsToWait) {
		requestHandler.waitForResponses(requestIds, maxSecondsToWait);
	}

	public void closeOrderBook(int requestId, boolean smartDepth) {
		client.cancelMktDepth(requestId, smartDepth);
		requestHandler.closeBook(requestId, smartDepth);
	}

	public void closeAccountListener() {
		client.cancelPositions();
	}
}
